tests.aignostics.application.utils_test::test_retrieve_and_print_run_details_default_is_detailed

Stack Traces | 0.006s run time

mock_console = <MagicMock name='console' id='140038505405408'>

    @pytest.mark.unit
    @patch("aignostics.application._utils.console")
    def test_retrieve_and_print_run_details_default_is_detailed(mock_console: Mock) -> None:
        """Test that default mode (summarize=False) shows detailed output with artifacts."""
        terminated_at = datetime(2025, 1, 1, 13, 0, 0, tzinfo=UTC)
    
        run_data = _make_run_data(
            run_id="run-detailed-test",
            application_id="he-tme",
            version_number="1.0.0",
            state=RunState.TERMINATED,
            termination_reason=RunTerminationReason.ALL_ITEMS_PROCESSED,
            output=RunOutput.FULL,
            statistics=_make_statistics(item_count=1, item_succeeded_count=1),
            terminated_at=terminated_at,
        )
    
        item_result = _make_item_result(
            item_id="item-123",
            terminated_at=terminated_at,
            output_artifacts=[_make_artifact()],
        )
    
        mock_run = MagicMock()
        mock_run.details.return_value = run_data
        mock_run.results.return_value = [item_result]
    
        # Call without summarize parameter (default is False)
        retrieve_and_print_run_details(mock_run, hide_platform_queue_position=False)
    
        all_output = " ".join(str(call) for call in mock_console.print.call_args_list)
    
        # Verify detailed output shows "Run Details" not "Run Summary"
        assert "Run Details for run-detailed-test" in all_output
        # Verify artifact details ARE shown in detailed mode
>       assert "Download URL" in all_output
E       assert 'Download URL' in "call('[bold]Run Details for run-detailed-test[/bold]\\n================================================================================\\n[bold]Run ID:[/bold] run-detailed-test\\n[bold]Application (Version):[/bold] he-tme (1.0.0)\\n[bold]Queue Position:[/bold] N/A\\n[bold]Status (Termination Reason):[/bold] TERMINATED (RunTerminationReason.ALL_ITEMS_PROCESSED)\\n[bold]Output:[/bold] FULL\\n[bold]Statistics:[/bold]\\n  - 1 items\\n  - 0 pending\\n  - 0 processing\\n  - 0 skipped\\n  - 1 succeeded\\n  - 0 user errors\\n  - 0 system errors\\n[bold]Submitted (by):[/bold] 2025-01-01 12:00:00+00:00 (user@example.com)\\n[bold]Terminated (duration):[/bold] 2025-01-01 13:00:00+00:00 (1 hour)\\n[bold]Custom Metadata:[/bold] None\\n\\n[bold]Items:[/bold]') call('  [bold]Item ID:[/bold] item-123\\n  [bold]Item External ID:[/bold] `slide-001.svs`\\n  [bold]Status (Termination Reason):[/bold] TERMINATED (ItemTerminationReason.SUCCEEDED)\\n  [bold]Error Message (Code):[/bold] None (None)\\n  [bold]Custom Metadata:[/bold] None\\n  [bold]Output Artifacts:[/bold]\\n    - Name: result.parquet\\n      MIME Type: application/vnd.apache.parquet\\n      Artifact ID: artifact-abc\\n')"

.../aignostics/application/utils_test.py:903: AssertionError

tests.aignostics.qupath.gui_test::test_gui_run_qupath_install_to_inspect

Stack Traces | 318s run time

user = <nicegui.testing.user.User object at 0x7f23181b3f00>
runner = <typer.testing.CliRunner object at 0x7f2320854130>
tmp_path = PosixPath('.../pytest-of-runner/pytest-21/test_gui_run_qupath_install_to0')
silent_logging = None, qupath_teardown = None, qupath_save_restore = None
record_property = <function record_property.<locals>.append_property at 0x7f231b1a1b10>

    @pytest.mark.e2e
    @pytest.mark.long_running
    @pytest.mark.skipif(
        (platform.system() == "Linux" and platform.machine() in {"aarch64", "arm64"}),
        reason="QuPath is not supported on ARM64 Linux",
    )
    @pytest.mark.timeout(timeout=60 * 15)
    @pytest.mark.sequential
    async def test_gui_run_qupath_install_to_inspect(  # noqa: C901, PLR0912, PLR0913, PLR0914, PLR0915, PLR0917
        user: User,
        runner: CliRunner,
        tmp_path: Path,
        silent_logging: None,
        qupath_teardown: None,
        qupath_save_restore: None,
        record_property,
    ) -> None:
        """Test installing QuPath, downloading run results, creating QuPath project from it, and inspecting results."""
        record_property("tested-item-id", "TC-QUPATH-01, SPEC-GUI-SERVICE")
    
        # Find run
        runs = Service().application_runs(
            application_id=HETA_APPLICATION_ID,
            application_version=HETA_APPLICATION_VERSION,
            external_id=SPOT_0_GS_URL,
            tags=["scheduled"],
            has_output=True,
            limit=1,
        )
        if not runs:
            message = f"No matching runs found for application {HETA_APPLICATION_ID} ({HETA_APPLICATION_VERSION}). "
            message += "This test requires the scheduled test test_application_runs_heta_version passing first."
            pytest.skip(message)
    
        run_id = runs[0].run_id
    
        # Explore run
        run = Service().application_run(run_id).details()
        print(
            f"Found existing run: {run.run_id}\n"
            f"application: {run.application_id} ({run.version_number})\n"
            f"status: {run.state}, output: {run.output}\n"
            f"submitted at: {run.submitted_at}, terminated at: {run.terminated_at}\n"
            f"statistics: {run.statistics!r}\n",
            f"custom_metadata: {run.custom_metadata!r}\n",
        )
    
        # Explore results
        results = list(Service().application_run(run_id).results())
        assert results, f"No results found for run {run_id}"
        for item in results:
            print(
                f"Found item: {item.item_id}, status: {item.state}, output: {item.output}, "
                f"external_id: {item.external_id}\n"
                f"custom_metadata: {item.custom_metadata!r}\n",
            )
    
        with patch(
            "aignostics.application._gui._page_application_run_describe.get_user_data_directory", return_value=tmp_path
        ):
            # Step 1: (Re)Install QuPath
            result = runner.invoke(cli, ["qupath", "install"])
            output = normalize_output(result.output, strip_ansi=True)
            assert f"QuPath v{QUPATH_VERSION} installed successfully" in output, (
                f"Expected 'QuPath v{QUPATH_VERSION} installed successfully' in output.\nOutput: {output}"
            )
            assert result.exit_code == 0
    
            # Step 2: Go to latest completed run via GUI
            await user.open(f"/application/run/{run.run_id}")
            await user.should_see(f"Run {run.run_id}")
            await user.should_see(f"Run of {HETA_APPLICATION_ID} ({HETA_APPLICATION_VERSION})")
    
            # Step 3: Open Result Download dialog
            await user.should_see(marker="BUTTON_OPEN_QUPATH", retries=100)
            user.find(marker="BUTTON_OPEN_QUPATH").click()
    
            # Step 4: Select Data destination
            await user.should_see(marker="BUTTON_DOWNLOAD_DESTINATION_DATA")
            download_destination_data_button: ui.button = user.find(
                marker="BUTTON_DOWNLOAD_DESTINATION_DATA"
            ).elements.pop()
            assert download_destination_data_button.enabled, "Download destination button should be enabled"
            user.find(marker="BUTTON_DOWNLOAD_DESTINATION_DATA").click()
            await assert_notified(user, "Using Launchpad results directory", 30)
    
            # Step 5: Trigger Download
            await user.should_see(marker="DIALOG_BUTTON_DOWNLOAD_RUN")
            download_run_button: ui.button = user.find(marker="DIALOG_BUTTON_DOWNLOAD_RUN").elements.pop()
            assert download_run_button.enabled, "Download button should be enabled before downloading"
            user.find(marker="DIALOG_BUTTON_DOWNLOAD_RUN").click()
            await assert_notified(user, "Downloading ...", 30)
    
            # Step 6: Check download completes, QuPath project created, and QuPath launched
>           await assert_notified(user, "Download and QuPath project creation completed.", 60 * 5)

.../aignostics/qupath/gui_test.py:232: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

user = <nicegui.testing.user.User object at 0x7f23181b3f00>
expected_notification = 'Download and QuPath project creation completed.'
wait_seconds = 300

    async def assert_notified(user: User, expected_notification: str, wait_seconds: int = 5) -> str:
        """Check if the user receives a notification within the specified time.
    
        This utility function helps test GUI notifications by waiting for a specific
        notification message to appear in the user's notification messages.
    
        Args:
            user: The nicegui User instance for testing.
            expected_notification: The notification text to look for (partial match).
            wait_seconds: Maximum time to wait for the notification (default: 5).
    
        Returns:
            str: The oldest matching notification message found.
    
        Raises:
            pytest.fail: If no matching notification is found within the wait time.
        """
        for _ in range(wait_seconds):
            matching_messages = [msg for msg in user.notify.messages if expected_notification in msg]
            if matching_messages:
                return matching_messages[0]
            await sleep(1)
    
        recent_messages = (user.notify.messages[-10:] if len(user.notify.messages) > 10 else user.notify.messages)[::-1]
        total_count = len(user.notify.messages)
>       pytest.fail(
            f"No notification containing '{expected_notification}' was found within {wait_seconds} seconds. "
            f"Total messages: {total_count}. Recent messages: {recent_messages}"
        )
E       Failed: No notification containing 'Download and QuPath project creation completed.' was found within 300 seconds. Total messages: 5. Recent messages: ['#x1F389 Run he-tme completed!', '#x1F389 Run he-tme completed!', '#x1F389 Run he-tme completed!', 'Downloading ...', 'Using Launchpad results directory']

tests/conftest.py:131: Failed

tests.aignostics.application.gui_test::test_gui_run_download

Stack Traces | 488s run time

user = <nicegui.testing.user.User object at 0x7f2353405810>
runner = <typer.testing.CliRunner object at 0x7f2353406780>
tmp_path = PosixPath('.../pytest-of-runner/pytest-21/test_gui_run_download1')
silent_logging = None
record_property = <function record_property.<locals>.append_property at 0x7f231bcb9220>

    @pytest.mark.e2e
    @pytest.mark.long_running
    @pytest.mark.flaky(retries=1, delay=5)
    @pytest.mark.timeout(timeout=60 * 10)
    @pytest.mark.sequential  # Helps on Linux with image analysis step otherwise timing out
    async def test_gui_run_download(  # noqa: PLR0915
        user: User, runner: CliRunner, tmp_path: Path, silent_logging: None, record_property
    ) -> None:
        """Test that the user can download a run result via the GUI."""
        record_property("tested-item-id", "SPEC-APPLICATION-SERVICE, SPEC-GUI-SERVICE")
        with patch(
            "aignostics.application._gui._page_application_run_describe.get_user_data_directory",
            return_value=tmp_path,
        ):
            # Find run
            runs = Service().application_runs(
                application_id=HETA_APPLICATION_ID,
                application_version=HETA_APPLICATION_VERSION,
                external_id=SPOT_0_GS_URL,
                tags=["scheduled"],
                has_output=True,
                limit=1,
            )
            if not runs:
                message = f"No matching runs found for application {HETA_APPLICATION_ID} ({HETA_APPLICATION_VERSION}). "
                message += "This test requires the scheduled test test_application_runs_heta_version passing first."
                pytest.skip(message)
    
            run_id = runs[0].run_id
    
            # Explore run
            run = Service().application_run(run_id).details()
            print(
                f"Found existing run: {run.run_id}\n"
                f"application: {run.application_id} ({run.version_number})\n"
                f"status: {run.state}, output: {run.output}\n"
                f"submitted at: {run.submitted_at}, terminated at: {run.terminated_at}\n"
                f"statistics: {run.statistics!r}\n",
                f"custom_metadata: {run.custom_metadata!r}\n",
            )
            # Step 1: Go to latest completed run
            await user.open(f"/application/run/{run.run_id}")
            await user.should_see(f"Run {run.run_id}", retries=100)
            await user.should_see(
                f"Run of {run.application_id} ({run.version_number})",
                retries=100,
            )
    
            # Step 2: Open Result Download dialog
            await user.should_see(marker="BUTTON_DOWNLOAD_RUN", retries=100)
            user.find(marker="BUTTON_DOWNLOAD_RUN").click()
    
            # Step 3: Check download button is initially disabled, then select Data folder
            download_run_button: ui.button = user.find(marker="DIALOG_BUTTON_DOWNLOAD_RUN").elements.pop()
            assert not download_run_button.enabled, "Download button should be disabled before selecting target"
            await user.should_see(marker="BUTTON_DOWNLOAD_DESTINATION_DATA", retries=100)
            user.find(marker="BUTTON_DOWNLOAD_DESTINATION_DATA").click()
            await assert_notified(user, "Using Launchpad results directory")
    
            # Step 4: Trigger Download - wait for button to be enabled
            download_run_button = user.find(marker="DIALOG_BUTTON_DOWNLOAD_RUN").elements.pop()
            assert download_run_button.enabled, "Download button should be enabled after selecting target"
            user.find(marker="DIALOG_BUTTON_DOWNLOAD_RUN").click()
            await assert_notified(user, "Downloading ...")
    
            # Check: Download completed
>           await assert_notified(user, "Download completed.", 60 * 4)

.../aignostics/application/gui_test.py:411: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

user = <nicegui.testing.user.User object at 0x7f2353405810>
expected_notification = 'Download completed.', wait_seconds = 240

    async def assert_notified(user: User, expected_notification: str, wait_seconds: int = 5) -> str:
        """Check if the user receives a notification within the specified time.
    
        This utility function helps test GUI notifications by waiting for a specific
        notification message to appear in the user's notification messages.
    
        Args:
            user: The nicegui User instance for testing.
            expected_notification: The notification text to look for (partial match).
            wait_seconds: Maximum time to wait for the notification (default: 5).
    
        Returns:
            str: The oldest matching notification message found.
    
        Raises:
            pytest.fail: If no matching notification is found within the wait time.
        """
        for _ in range(wait_seconds):
            matching_messages = [msg for msg in user.notify.messages if expected_notification in msg]
            if matching_messages:
                return matching_messages[0]
            await sleep(1)
    
        recent_messages = (user.notify.messages[-10:] if len(user.notify.messages) > 10 else user.notify.messages)[::-1]
        total_count = len(user.notify.messages)
>       pytest.fail(
            f"No notification containing '{expected_notification}' was found within {wait_seconds} seconds. "
            f"Total messages: {total_count}. Recent messages: {recent_messages}"
        )
E       Failed: No notification containing 'Download completed.' was found within 240 seconds. Total messages: 5. Recent messages: ['#x1F389 Run he-tme completed!', '#x1F389 Run he-tme completed!', '#x1F389 Run he-tme completed!', 'Downloading ...', 'Using Launchpad results directory']

tests/conftest.py:131: Failed